package com.gpj.netty.application.server;

import com.sun.corba.se.impl.orbutil.LogKeywords;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * @className SelectorServer
 * @description 使用多路复用的NIO服务端
 *   * basicReactorDesign -- 最基本的Reactor设计
 *   * workerThreadPool -- 使用工作线程池的Reactor设计
 *   * multipleReactors -- 多reactor + 工作线程池 的Reactor设计模式
 * @author GPJ
 * @date 2021/11/30 19:39
 * @version 1.0
 **/
@Slf4j
public class NIOSelectorServer {
	private static ThreadPoolExecutor executor = new ThreadPoolExecutor(2,
			5, 1, TimeUnit.SECONDS, new LinkedBlockingQueue<>());
	private static Map<SelectionKey, SelectionKey> handing = new ConcurrentHashMap<>();

	public static void main(String[] args) throws IOException {
//		NIOSelectorServer.basicReactorDesign();
//		NIOSelectorServer.workerThreadPool();
		NIOSelectorServer.multipleReactors();
	}

	/**
	 * 最基本的Reactor设计
	 */
	public static void basicReactorDesign() throws IOException {
		log.debug("服务端开始启动...");
		ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
		serverSocketChannel.socket().bind(new InetSocketAddress(9000));

		// 设置ServerSocketChannel为非阻塞
		serverSocketChannel.configureBlocking(false);

		// 打开Selector处理Channel，即创建epoll
		Selector selector = Selector.open();

		// 把ServerSocketChannel注册到selector上，并且selector对客户端accept连接操作感兴趣
		serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
		log.debug("服务端启动成功");

		while (true){
			// 阻塞等待需要处理的事件发生
			selector.select();

			Set<SelectionKey> keys = selector.selectedKeys();
			Iterator<SelectionKey> iterator = keys.iterator();
			while (iterator.hasNext()) {
				SelectionKey key = iterator.next();
				if (key.isAcceptable()) {
					acceptHandle(selector,key);
				} else if (key.isReadable()) {
					readHandle(key);
				} else if (key.isWritable()) {
					writeHandle(key);
				}
				// 从事件集合里删除本次处理的key，防止下次select重复处理
				iterator.remove();
			}
		}
	}

	/**
	 * 使用工作线程池的Reactor设计
	 */
	public static void workerThreadPool() throws IOException {
		log.debug("服务端开始启动...");
		ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
		serverSocketChannel.socket().bind(new InetSocketAddress(9000));

		// 设置ServerSocketChannel为非阻塞
		serverSocketChannel.configureBlocking(false);

		// 打开Selector处理Channel，即创建epoll
		Selector selector = Selector.open();

		// 把ServerSocketChannel注册到selector上，并且selector对客户端accept连接操作感兴趣
		serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
		log.debug("服务端启动成功");

		while (true){
			// 阻塞等待需要处理的事件发生
			selector.select(300);

			Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
			while (iterator.hasNext()) {
				SelectionKey key = iterator.next();

				if (key.isAcceptable()) {
					acceptHandle(selector, key);
				} else if(!handing.containsKey(key)){
					handing.put(key, key); // 防止重复操作(线程没有处理完成之后的重复操作)
					executor.execute(() -> {
						try {
							if (key.isReadable()) {
								readHandle(key);
							} else if (key.isWritable()){
								writeHandle(key);
							}
						} catch (IOException e) {
							log.error(e.toString(), e);
						} finally {
							handing.remove(key);
						}
					});
				}
				// 从事件集合里删除本次处理的key，防止下次select重复处理
				iterator.remove();
			}
		}
	}

	/**
	 * 多reactor + 工作线程池 的Reactor设计模式
	 */
	public static void multipleReactors() throws IOException {
		/******************** 服务端启动 ********************/
		log.debug("服务端开始启动...");
		ServerSocketChannel serverSocketChannel = ServerSocketChannel.open();
		serverSocketChannel.socket().bind(new InetSocketAddress(9000));

		// 设置ServerSocketChannel为非阻塞
		serverSocketChannel.configureBlocking(false);

		// 打开Selector处理Channel，即创建epoll
		Selector selector = Selector.open();

		// 把ServerSocketChannel注册到selector上，并且selector对客户端accept连接操作感兴趣
		serverSocketChannel.register(selector, SelectionKey.OP_ACCEPT);
		log.debug("服务端启动成功");

		/******************** 启动read/writ处理的selector线程 ********************/
		// 构建read/writ的reactor线程
		Selector subSelector = Selector.open();
		new Thread(() -> {
			while (true){
				try {
					subSelector.select(300);
				} catch (IOException e) {
					log.error(e.toString(), e);
					continue;
				}

				Iterator<SelectionKey> iterator = subSelector.selectedKeys().iterator();
				while (iterator.hasNext()) {
					SelectionKey key = iterator.next();
					if(!handing.containsKey(key)){
						handing.put(key, key); // 防止重复操作(线程没有处理完成之后的重复操作)
						executor.execute(() -> {
							try {
								if (key.isReadable()) {
									readHandle(key);
								} else if (key.isWritable()){
									writeHandle(key);
								}
							} catch (IOException e) {
								log.error(e.toString(), e);
							} finally {
								handing.remove(key);
							}
						});
					}
					iterator.remove();
				}
			}
		}).start();

		/******************** 启动accept处理的selector线程 ********************/
		while (true){
			selector.select();

			Iterator<SelectionKey> iterator = selector.selectedKeys().iterator();
			while (iterator.hasNext()) {
				SelectionKey key = iterator.next();
				if (key.isAcceptable()) acceptHandle(subSelector, key);
				iterator.remove();
			}
		}
	}

	private static void acceptHandle(Selector selector, SelectionKey key) throws IOException {
		SocketChannel socketChannel = ((ServerSocketChannel) key.channel()).accept();
		socketChannel.configureBlocking(false);
		// 注册读事件
		socketChannel.register(selector, SelectionKey.OP_READ);
		log.debug("[{}]客户端连接成功", socketChannel.getRemoteAddress());
	}
	
	private static void readHandle(SelectionKey key) throws IOException {
		SocketChannel socketChannel = (SocketChannel) key.channel();
		ByteBuffer buffer = ByteBuffer.allocate(1024);
		int len = socketChannel.read(buffer);

		if (len > 0) {
			log.info("[{}]客户端信息：{}", socketChannel.getRemoteAddress(), new String(buffer.array(), 0, len));

			log.debug("[{}]客户端请求接受完毕", socketChannel.getRemoteAddress());

			log.debug("开始响应请求...");
			ByteBuffer writeBuffer = ByteBuffer.wrap("请求请求接收完成".getBytes());
			socketChannel.write(writeBuffer);
			key.interestOps(SelectionKey.OP_READ | SelectionKey.OP_WRITE);
		}
	}
	
	private static void writeHandle(SelectionKey key) throws IOException {
		log.debug("响应请求完毕");
		// NIO事件触发是水平触发,使用Java的NIO编程的时候，在没有数据可以往外写的时候要取消写事件，
		key.interestOps(SelectionKey.OP_READ);
	}
}
